//package com.boot.comsumer;
//
//import com.alibaba.fastjson.JSONObject;
//import com.boot.entity.Config;
//import com.boot.entity.config_canal.ConfigCanalBean;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.kafka.clients.consumer.ConsumerRecord;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.data.redis.core.RedisTemplate;
//import org.springframework.kafka.annotation.KafkaListener;
//import org.springframework.kafka.support.Acknowledgment;
//import org.springframework.stereotype.Component;
//
//import java.util.List;
//import java.util.concurrent.TimeUnit;
//
///**
// * kafka消费者（监听名为canal-test-topic的topic），同步Redis
// * @author youzhengjie 2022-09-01 16:54:28
// */
//@Component
//@Slf4j
//public class ConfigCanalRedisConsumer {
//
//    @Autowired
//    private RedisTemplate redisTemplate;
//
//    //redis的key格式：（数据库.表名_字段的id）
//    private static final String KEY_PREFIX = "canal-test-db1.t_config_";
//
//    //过期时间（单位：小时）
//    private static final int TIME_OUT = 24;
//
//
//    /**
//     * @param consumer 接收消费记录（消息）
//     * @param ack 手动提交消息
//     */
//    @KafkaListener(topics = "canal-test-topic")
//    public void receive(ConsumerRecord<String, String> consumer, Acknowledgment ack) {
//
//        try {
//            //获取canal的消息
//            String value = (String) consumer.value();
//            log.info("topic名称:{},key:{},分区位置:{},下标:{},value:{}", consumer.topic(), consumer.key(), consumer.partition(), consumer.offset(), value);
//
//            //转换为javaBean
//            ConfigCanalBean canalBean = JSONObject.parseObject(value, ConfigCanalBean.class);
//            /*
//             由于我们canal.instance配置了监控canal-test-db1.t_config表和canal-test-db1.t_user表（生产环境下可以启动多个canal，每一个canal监听一张需要同步的表）
//            所以我们要对这两张表分开处理。（可以通过他们的表名（canalBean.getTable()）来区分）
//            如果canalBean.getTable()获取的表名是t_config，则同步到redis，如果不是则不管。
//             */
//            //System.out.println(canalBean);
//            if("t_config".equals(canalBean.getTable())){
//                //获取是否是DDL语句
//                boolean isDdl = canalBean.isDdl();
//                //获取当前sql语句的类型（比如INSERT、DELETE等等）
//                String type = canalBean.getType();
//                List<Config> configList = canalBean.getData();
//                //如果不是DDL语句
//                if (!isDdl) {
//                    //INSERT和UPDATE都是一样的操作
//                    if ("INSERT".equals(type) || "UPDATE".equals(type)) {
//                        //新增语句
//                        for (Config config : configList) {
//                            Long id = config.getConfigId();
//                            //新增到redis中,过期时间是10分钟
//                            redisTemplate.
//                                    opsForValue().
//                                    set(KEY_PREFIX + id, JSONObject.toJSONString(config), TIME_OUT, TimeUnit.HOURS);
//                        }
//                    }else if("DELETE".equals(type)){
//                        //删除语句
//                        for (Config config : configList) {
//                            Long id = config.getConfigId();
//                            //从redis中删除
//                            redisTemplate.delete(KEY_PREFIX+id);
//                        }
//                    }
//
//                }
//            }
//            //最后，如果上面的代码没有报错的情况下，可以确认消息了。（很重要）
//            ack.acknowledge();
//        }catch (Exception e){
//            throw new RuntimeException();
//        }
//    }
//
//}
//
